/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.sdk.io;



import org.apache.avro.Schema;

import org.apache.avro.file.CodecFactory;

import com.bff.gaia.unified.sdk.io.FileBasedSink.FilenamePolicy;

import com.bff.gaia.unified.sdk.transforms.SerializableFunction;

import com.bff.gaia.unified.sdk.transforms.display.DisplayData;

import com.bff.gaia.unified.sdk.transforms.display.HasDisplayData;

import com.bff.gaia.unified.vendor.guava.com.google.common.base.Function;

import com.bff.gaia.unified.vendor.guava.com.google.common.base.Supplier;

import com.bff.gaia.unified.vendor.guava.com.google.common.base.Suppliers;

import com.bff.gaia.unified.vendor.guava.com.google.common.io.BaseEncoding;



import javax.annotation.Nullable;

import java.io.Serializable;

import java.util.Map;



/** Always returns a constant {@link FilenamePolicy}, {@link Schema}, metadata, and codec. */

class ConstantAvroDestination<UserT, OutputT>

    extends DynamicAvroDestinations<UserT, Void, OutputT> {

  private static class SchemaFunction implements Serializable, Function<String, Schema> {

    @Nullable

    @Override

    public Schema apply(@Nullable String input) {

      return new Schema.Parser().parse(input);

    }

  }



  // This should be a multiple of 4 to not get a partial encoded byte.

  private static final int METADATA_BYTES_MAX_LENGTH = 40;

  private final FilenamePolicy filenamePolicy;

  private final Supplier<Schema> schema;

  private final Map<String, Object> metadata;

  private final SerializableAvroCodecFactory codec;

  private final SerializableFunction<UserT, OutputT> formatFunction;



  private class Metadata implements HasDisplayData {

    @Override

    public void populateDisplayData(DisplayData.Builder builder) {

      for (Map.Entry<String, Object> entry : metadata.entrySet()) {

        DisplayData.Type type = DisplayData.inferType(entry.getValue());

        if (type != null) {

          builder.add(DisplayData.item(entry.getKey(), type, entry.getValue()));

        } else {

          String base64 = BaseEncoding.base64().encode((byte[]) entry.getValue());

          String repr =

              base64.length() <= METADATA_BYTES_MAX_LENGTH

                  ? base64

                  : base64.substring(0, METADATA_BYTES_MAX_LENGTH) + "...";

          builder.add(DisplayData.item(entry.getKey(), repr));

        }

      }

    }

  }



  public ConstantAvroDestination(

      FilenamePolicy filenamePolicy,

      Schema schema,

      Map<String, Object> metadata,

      CodecFactory codec,

      SerializableFunction<UserT, OutputT> formatFunction) {

    this.filenamePolicy = filenamePolicy;

    this.schema = Suppliers.compose(new SchemaFunction(), Suppliers.ofInstance(schema.toString()));

    this.metadata = metadata;

    this.codec = new SerializableAvroCodecFactory(codec);

    this.formatFunction = formatFunction;

  }



  @Override

  public OutputT formatRecord(UserT record) {

    return formatFunction.apply(record);

  }



  @Override

  @Nullable

  public Void getDestination(UserT element) {

    return (Void) null;

  }



  @Override

  @Nullable

  public Void getDefaultDestination() {

    return (Void) null;

  }



  @Override

  public FilenamePolicy getFilenamePolicy(Void destination) {

    return filenamePolicy;

  }



  @Override

  public Schema getSchema(Void destination) {

    return schema.get();

  }



  @Override

  public Map<String, Object> getMetadata(Void destination) {

    return metadata;

  }



  @Override

  public CodecFactory getCodec(Void destination) {

    return codec.getCodec();

  }



  @Override

  public void populateDisplayData(DisplayData.Builder builder) {

    filenamePolicy.populateDisplayData(builder);

    builder.add(DisplayData.item("schema", schema.get().toString()).withLabel("Record Schema"));

    builder.addIfNotDefault(

        DisplayData.item("codec", codec.getCodec().toString()).withLabel("Avro Compression Codec"),

        AvroIO.TypedWrite.DEFAULT_SERIALIZABLE_CODEC.toString());

    builder.include("Metadata", new Metadata());

  }

}